package com.hrt.iceberg.unauto

import org.apache.hadoop.conf.Configuration
import org.apache.iceberg.Table
import org.apache.iceberg.actions.Actions
import org.apache.iceberg.catalog.TableIdentifier
import org.apache.iceberg.hadoop.HadoopCatalog
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * spark与iceberg整合查询操作
 * 1.加载表数据
 * 2.查询快照 历史信息 数据文件等信息
 * 3.根据指定快照查询数据
 * 4.根据指定时间戳查询数据
 * 5.指定快照的回滚
 * 6.合并iceberg的小文件
 * 7.删除iceberg的历史快照
 */
object SparkIceberg3 {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder().master("local").appName("SparkOperateIceberg")
      //设置Hadoop Catalog
      .config("spark.sql.catalog.hadoop_prod", "org.apache.iceberg.spark.SparkCatalog")
      .config("spark.sql.catalog.hadoop_prod.type", "hadoop")
      .config("spark.sql.catalog.hadoop_prod.warehouse", "hdfs://mycluster/sparkoperateiceberg")
      .config("spark.sql.extensions","org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
      .getOrCreate()



    /**
      *  删除 Iceberg 历史快照
      */
    val conf = new Configuration()
    val catalog = new HadoopCatalog(conf,"hdfs://mycluster/sparkoperateiceberg")
    val table: Table = catalog.loadTable(TableIdentifier.of("mydb","mytest"))
    table.expireSnapshots().expireOlderThan(1640677739793L).commit() // 早于当前时间戳的快照都会被删掉

    /**
      *  合并 Iceberg表小文件
      */
    //读取数据文件加载成DataFrame ,保存到iceberg表中
 /*   import spark.implicits._
    val df: DataFrame = spark.read.textFile("F:\\P8\\workspace\\bigdatalake\\icebergdata\\nameinfo")
      .map(line => {
        val arr: Array[String] = line.split(",")
        (arr(0).toInt, arr(1), arr(2).toInt)
      }).toDF("id", "name", "age")*/
//
//    df.writeTo("hadoop_prod.mydb.mytest").append()

//    spark.sql("select * from hadoop_prod.mydb.mytest").show(300)

    //java api 合并iceberg 文件
//    val conf = new Configuration()
//    val catalog = new HadoopCatalog(conf,"hdfs://hadoop102:8020/sparkoperateiceberg")
//    val table: Table = catalog.loadTable(TableIdentifier.of("mydb","mytest"))
//    Actions.forTable(table).rewriteDataFiles().targetSizeInBytes(1024)//1kb，指定生成合并之后文件大小
//    .execute()



    /**
      *  Spark 回滚快照
      *
      */
//    spark.sql(
//      """
//        |drop table hadoop_prod.mydb.mytest
//      """.stripMargin)
//
//    spark.sql(
//      """
//        |create table hadoop_prod.mydb.mytest (id int,name string,age int) using iceberg
//      """.stripMargin)
//    spark.sql(
//      """
//        |insert into hadoop_prod.mydb.mytest values (1,"zs",18),(2,"ls",19),(3,"ww",20)
//      """.stripMargin)

//    spark.sql(
//      """
//        |insert into hadoop_prod.mydb.mytest values (4,"ml",18),(5,"tq",19),(6,"gb",20)
//      """.stripMargin)

    /**回滚快照 - java api 方式*/
//    val conf = new Configuration()
//    val catalog = new HadoopCatalog(conf,"hdfs://hadoop102:8020/sparkoperateiceberg")
//    catalog.setConf(conf)
//    val table: Table = catalog.loadTable(TableIdentifier.of("mydb","mytest"))
//    table.manageSnapshots().rollbackTo(4428163433183700863L).commit() // 回滚到指定快照

//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest
//      """.stripMargin).show()

    /**回滚快照 -SQL 方式*/
//    spark.sql(
//      """
//        |call hadoop_prod.system.rollback_to_snapshot('mydb.mytest',4428163433183700863)
//      """.stripMargin)




    /**
      *  Spark 基于指定时间戳查询iceberg表数据
      */
//    spark.read
//      .option("as-of-timestamp","1640672100000")
//      .format("iceberg")
//      .load("hdfs://mycluster/sparkoperateiceberg/mydb/mytest")
//      .show()

    // sql的方式是spark3.x版本支持的 第一步相当于是设置了时间，第二步是查询设置的时间数据
//    spark.sql(
//      """
//        | call hadoop_prod.system.rollback_to_timestamp("mydb.mytest",TIMESTAMP '2021-12-28 14:15:00.000')
//      """.stripMargin)
//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest
//      """.stripMargin).show()


    /**
      *  SparkSQL 读取 指定快照数据
      */
//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest
//      """.stripMargin).show()

    //读取指定快照数据
//    spark.read
//      .option("snapshot-id",5725035405694513291L)
//      .format("iceberg")
//      .load("hdfs://hadoop102:8020/sparkoperateiceberg/mydb/mytest")
//      .show()

    // sql的方式是spark3.x版本支持的 第一步相当于是设置了快照，第二步是查询设置的快照
//    spark.sql(
//      """
//        | call hadoop_prod.system.set_current_snapshot('mydb.mytest',2722407523787742648)
//      """.stripMargin)
//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest
//      """.stripMargin).show()


    /**
      * 获取Iceberg表快照信息
      */
//    spark.sql(
//      """
//        |create table hadoop_prod.mydb.mytest (id int,name string,age int) using iceberg
//      """.stripMargin)
//    spark.sql(
//      """
//        |insert into hadoop_prod.mydb.mytest values (1,"zs",18),(2,"ls",19),(3,"ww",20)
//      """.stripMargin)
//    spark.sql(
//      """
//        |insert into hadoop_prod.mydb.mytest values (4,"ml",18),(5,"tq",19),(6,"gb",20)
//      """.stripMargin)

    //使用SparkSQL查询iceberg 表快照信息
//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest.snapshots
//      """.stripMargin).show()
    //使用SparkSQL查询iceberg 表历史
//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest.history
//      """.stripMargin).show()
    //使用SparkSQL查询iceberg datafile
//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest.files
//      """.stripMargin).show()
    //使用SparkSQL查询iceberg manifests
//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest.manifests
//      """.stripMargin).show()


    /**
      *  Spark DataFrame 加载Iceberg表中数据
      *  1) spark.read.format("iceberg").load("iceberg table path")
      *  2) spark.table("iceberg tablename")
      */
//    val df: DataFrame = spark.read.format("iceberg").load("hdfs://hadoop102:8020/sparkoperateiceberg/default/mytbl")
//    df.show()
//    val df2: DataFrame = spark.table("hadoop_prod.default.mytbl")
//    df2.show()



  }

}
